package com.study.hadoop.weather;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * 找到每月温度最高的两天
 * @author stephen
 * @create 2019-01-10 20:15
 * @since 1.0.0
 */
public class WeatherApp {

    /**
     * hdfs://localhost:9000/test/weather/in/weather.data hdfs://localhost:9000/test/weather/out
     * @param args
     * @throws IOException
     * @throws ClassNotFoundException
     * @throws InterruptedException
     */
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        if(args.length <2){
            throw new IllegalArgumentException("Usage: <input path> <output path>");
        }

        Configuration conf = new Configuration() ;
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
        conf.set("mapreduce.framework.name", "local");
        Job job = Job.getInstance(conf,"Weather App");

        job.setJarByClass(WeatherApp.class);

        //------------- start set mapper -----------
        //job.setInputFormatClass(WeatherInputFormat.class);
        job.setMapperClass(WeatherMapper.class);
        job.setMapOutputKeyClass(Weather.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setPartitionerClass(WeatherPartitioner.class);
        job.setSortComparatorClass(WeatherSortComparator.class);
        //job.setCombinerClass(WeatherCombiner.class);
        //------------- end set mapper -------------

        //------------- start set reducer -----------
        job.setGroupingComparatorClass(WeatherGroupComparator.class);
        job.setReducerClass(WeatherReducer.class);
        //------------- end set reducer -------------

        Path input = new Path(args[0]) ;
        FileInputFormat.addInputPath(job,input);

        Path output = new Path(args[1]) ;
        if(output.getFileSystem(conf).exists(output)){
            output.getFileSystem(conf).delete(output,true);
        }
        FileOutputFormat.setOutputPath(job, output);

        job.setNumReduceTasks(2);

        job.waitForCompletion(true) ;

    }
}
